package org.etlt.extract;

import org.etlt.expression.VariableContext;
import org.etlt.expression.datameta.Variable;
import org.etlt.job.JobContext;
import org.etlt.job.PkRange;
import org.etlt.load.ColumnSetting;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class LinkedExtractor extends Extractor {

    private final LinkedExtractorSetting setting;

    public LinkedExtractor(LinkedExtractorSetting setting) {
        this.setting = setting;
        setName(this.setting.getName());
    }

    private transient boolean moveLeft = true;

    private transient boolean moveRight = true;

    private boolean moveLeft(String sorting, boolean leftIsGreater) {
        if (sorting.equals("asc") || sorting.equals("desc")) {
            return sorting.equals("asc") ^ leftIsGreater;
        }
        throw new IllegalArgumentException("unsupported sorting type: " + sorting);
    }

    @Override
    public ExtractorSetting getSetting() {
        return this.setting;
    }

    @Override
    public void extract(JobContext context) {
        if (this.setting.getType().equals("join"))
            extract_join(context);
        else if (this.setting.getType().equals("left_join"))
            extract_left_join(context);
        else
            throw new IllegalArgumentException("unsupported linking type: " + this.setting.getType());
    }

    public void extract_join(JobContext context) {
        //-- support "join"
        if (!getLeftExtractor(context).isClosed() && !getRightExtractor(context).isClosed()) {
            getLeftExtractor(context).extract(context);
            getRightExtractor(context).extract(context);
            //using String to compare
            String leftValue = String.valueOf(context.getExpressionEvaluator().evaluate(this.setting.getCriteria().get(0), context));
            String rightValue = String.valueOf(context.getExpressionEvaluator().evaluate(this.setting.getCriteria().get(1), context));
            for (int compared; ; ) {
                compared = leftValue.compareTo(rightValue);
                if (compared == 0) {
                    Map<String, Object> result = new HashMap<>();
                    List<ColumnSetting> columns = this.setting.getColumnSettings();
                    columns.forEach((column) -> {
                        result.put(column.getName(), context.getExpressionEvaluator().evaluate(column.getExpression(), context));
                    });
                    Entity entity = new Entity(index++, result);
                    context.setEntity(this.setting.getName(), entity);

                    getLeftExtractor(context).extract(context);
                    getRightExtractor(context).extract(context);
                    break;
                }
                boolean isMoveLeft = moveLeft(setting.getSort(), compared > 0);
                if (isMoveLeft) {
                    getLeftExtractor(context).extract(context);
                } else {
                    getRightExtractor(context).extract(context);
                }
                if (!context.exists(getLeftExtractor(context).getName()) ||
                        !context.exists(getRightExtractor(context).getName())) {
                    context.removeEntity(getName());
                    break;
                }
            }
        } else {
            context.removeEntity(getName());
            close();
        }
    }

    public void extract_left_join(JobContext context) {
        //-- support "left_join"
        if (moveLeft && !getLeftExtractor(context).isClosed()) {
            getLeftExtractor(context).extract(context);
        }
        if (moveRight && !getRightExtractor(context).isClosed()) {
            getRightExtractor(context).extract(context);
        }
        Object left = context.getExpressionEvaluator().evaluate(this.setting.getCriteria().get(0), context);
        Object right = context.getExpressionEvaluator().evaluate(this.setting.getCriteria().get(1), context);
        if (left == null && right == null) {
            direct(false, false);
            context.removeEntity(getName());
            close();
        } else if (left == null && right != null) {
            // remove the left object and move right
            context.removeEntity(getLeftExtractor(context).getName());
            direct(false, true);
            buildEntity(context, context);
        } else if (right == null && left != null) {
            // remove the right object and move left
            context.removeEntity(getRightExtractor(context).getName());
            direct(true, false);
            buildEntity(context, context);
        } else {
            String leftValue = String.valueOf(left);
            String rightValue = String.valueOf(right);
            int compared = leftValue.compareTo(rightValue);
            if (compared == 0) {
                buildEntity(context, context);
                direct(true, true);
            } else {
                boolean isMoveLeft = moveLeft(setting.getSort(), compared > 0);
                VariableContext variableContext;
                if (isMoveLeft) {
                    direct(true, false);
                    variableContext = buildVariableContext(context, getLeftExtractor(context).getName());
                } else {
                    direct(false, true);
                    variableContext = buildVariableContext(context, getRightExtractor(context).getName());
                }
                buildEntity(context, variableContext);
            }
        }
    }

    private void direct(boolean left, boolean right) {
        this.moveLeft = left;
        this.moveRight = right;
    }

    private void buildEntity(JobContext context, VariableContext variableContext) {
        context.removeEntity(this.setting.getName());
        Map<String, Object> result = new HashMap<>();
        List<ColumnSetting> columns = this.setting.getColumnSettings();
        columns.forEach((column) -> {
            result.put(column.getName(), context.getExpressionEvaluator().evaluate(column.getExpression(), variableContext));
        });
        Entity entity = new Entity(index++, result);
        context.setEntity(this.setting.getName(), entity);
    }

    @Override
    public List<String> getColumns() {
        List<String> columns = new ArrayList<>();
        List<ColumnSetting> settings = this.setting.getColumnSettings();
        settings.forEach((setting) -> {
            columns.add(setting.getName());
        });
        return columns;
    }

    @Override
    public LinkedExtractor createInstance() {
        return new LinkedExtractor(this.setting);
    }

    @Override
    public void close() {
        this.closed = true;
    }

    @Override
    public boolean isClosed() {
        return false;
    }

    protected Extractor getLeftExtractor(JobContext context) {
        return context.getExtractor(this.setting.getExtractors().get(0));
    }

    protected Extractor getRightExtractor(JobContext context) {
        return context.getExtractor(this.setting.getExtractors().get(1));
    }

    private VariableContext buildVariableContext(JobContext context, String extractor) {
        return name -> {
            Entity entity = context.getEntity(extractor);
            if (entity != null) {
                int index = name.indexOf('.');
                if (index != -1) {
                    String entityName = name.substring(0, index);
                    String fieldName = name.substring(index + 1);
                    if (entityName.equals(extractor))
                        return Variable.createVariable(name, entity.getValue(fieldName));
                } else {
                    return Variable.createVariable(name, entity.getValue(name));
                }
            }
            return null;
        };
    }

    @Override
    public PkRange getPkFullRange() {
        throw new UnsupportedOperationException("not supported by FileExtractor.");
    }
}
